package joindemo.sse.util;

import joindemo.common.enums.HttpStatusEnum;
import joindemo.sse.SseUTF8;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Description SSE消息推送工具
 */
@Component
@Slf4j
public class SseUtils {

    /**
     * 容器
     */
    private static final ConcurrentHashMap<Long, SseUTF8> SSE_MAP_CACHE=new ConcurrentHashMap<>(0);

    public static SseUTF8 getSseUTF8(Long key){
        return SSE_MAP_CACHE.get(key);
    }

    /**
     * 默认时长不过期
     */
    private static final long DEFAULT_TIMEOUT=0L;

    /**
     * 创建连接
     * @param userId
     * @return SseUTF8
     */
    public static SseUTF8 createConnect(Long userId){
        SseUTF8 sseEmitter=new SseUTF8(DEFAULT_TIMEOUT);
        //需要给客户端推送id；
        if(SSE_MAP_CACHE.containsKey(userId)){
            log.info("SSE老用户连接:{}",userId);
            remove(userId);
        }
        //长连接完成后回调接口（关闭连接时调用）
        sseEmitter.onCompletion(()->{
            log.info("SSE连接结束:{}",userId);
            remove(userId);
        });
        //连接超时回调接口
        sseEmitter.onTimeout(()->{
            log.info("SSE连接超时:{}",userId);
            remove(userId);
        });
        //连接异常时，回调方法
        sseEmitter.onError(throwable -> {
            try{
                log.info("SSE{}连接异常，{}",userId,throwable.toString());
                sseEmitter.send(SseUTF8.event()
                        .id(String.valueOf(userId))
                        .name("发生异常！")
                        .data("发生异常重试！")
                        .reconnectTime(3000));
                SSE_MAP_CACHE.put(userId,sseEmitter);
            }catch(IOException e){
                log.error("用户--->{}SSE连接失败重试，异常信息--->{}",userId,e.getMessage());
                e.printStackTrace();
            }
        });
        SSE_MAP_CACHE.put(userId,sseEmitter);
        try{
            //注册成功返回用户信息
            sseEmitter.send(SseUTF8.event()
                    .id(String.valueOf(HttpStatusEnum.SUCCESS.getCode()))
                    .data(userId, MediaType.APPLICATION_JSON));
        }catch(IOException e){
            log.error("用户--->{} SSE连接失败，异常信息--->{}",userId,e.getMessage());
            e.printStackTrace();
        }
        return sseEmitter;
    }

    /**
     * 移除连接
     * @param userId
     * @return void
     */
    private static void remove(Long userId){
        //获取连接对象并关闭连接
        SSE_MAP_CACHE.get(userId).complete();
        SSE_MAP_CACHE.remove(userId);
        log.info("SSE移除用户连接---->{}",userId);
    }

    /**
     * 关闭连接
     * @param userId
     * @return void
     */
    public static void closeConnect(Long userId){
        log.info("SSE关闭连接:{}",userId);
        remove(userId);
    }

    /**
     * 发送消息给用户
     * @param userId
     * @param message
     * @param sseEmitter
     * @return boolean
     * @param <T>
     */
    private static <T>boolean sendToClient(Long userId, String message, SseUTF8 sseEmitter){
        //推送之前检测心跳是否存在
        boolean isAlive=checkSseConnectAlive(sseEmitter);
//        WSMessage.UserInfo fromUser=wsMessage.getFromUser();
//        WSMessage.Message message=wsMessage.getMessage();
        if(!isAlive){
            //失去连接移除
            log.error("SSE推送失败：客户端{}未创建长连接或者关闭, 失败信息:{}",userId,message);
            SSE_MAP_CACHE.remove(userId);
            return false;
        }
        SseEmitter.SseEventBuilder sendData=SseUTF8.event()
                .id(String.valueOf(HttpStatusEnum.SUCCESS.getCode()))
                .data(message,MediaType.APPLICATION_JSON);
        try {
            sseEmitter.send(sendData);
            return true;
        }catch(IOException e){
            log.error("SSE推送失败：用户{}，推送信息{}",userId,message);
        }
        return true;
    }

    /**
     * 检查连接是否还存活
     * @param sseEmitter
     * @return
     */
    public static boolean checkSseConnectAlive(SseUTF8 sseEmitter){
        if(sseEmitter==null){
            return false;
        }
        //返回true代表还连接，返回false代表失去连接
        return !(Boolean)getField(sseEmitter,sseEmitter.getClass(),"sendFailed")&&
                !(Boolean)getField(sseEmitter,sseEmitter.getClass(),"complete");
    }

    /**
     * 获取对象属性
     * @param obj
     * @param clazz
     * @param fieldName
     * @return
     */
    public static Object getField(Object obj, Class<?> clazz, String fieldName){
        for(; clazz != Object.class; clazz = clazz.getSuperclass()){
            try{
                Field field = clazz.getDeclaredField(fieldName);
                field.setAccessible(true);
                return field.get(obj);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
        return null;
    }

    /**
     * 发送消息给所有客户端
     * @param message
     * @return void
     * @param <T>
     */
    public static <T>boolean sendTextMessage(String message){
        if(SSE_MAP_CACHE.isEmpty()){
            return false;
        }
        if(message==null){
            return false;
        }
        boolean isSuccess=false;
        for(Map.Entry<Long,SseUTF8> entry:SSE_MAP_CACHE.entrySet()){
            isSuccess=sendToClient(entry.getKey(),message,entry.getValue());
            if(!isSuccess){
                log.error("群发客户端{}消息推送，失败消息{}",
                    entry.getKey(),
                    message);
            }
        }
        return isSuccess;
    }

    /**
     * 发送消息给指定用户
     * @param clientId
     * @param message
     * @return
     * @param <T>
     */
    public static <T>boolean sendTestMessage(Long clientId,String message){
        return sendToClient(clientId,message,SSE_MAP_CACHE.get(clientId));
    }

    /**
     * 发送消息检测心跳
     * @param live
     * @return void
     * @param <T>
     */
    public static <T>boolean sendTextMessageCheck(String live){
        if(SSE_MAP_CACHE.isEmpty()){
            return false;
        }
        if(live==null){
            return false;
        }
        boolean isSuccess=false;
        for(Map.Entry<Long,SseUTF8> entry:SSE_MAP_CACHE.entrySet()){
            isSuccess=sendToClientCheck(entry.getKey(), live,entry.getValue());
            if(!isSuccess){
                log.error("群发客户端{}消息推送，心跳检查，失败消息：{}",
                    entry.getKey(),
                    live);
            }
        }
        return isSuccess;
    }

    /**
     * 推送信息到客户端(检查)
     * @param userId
     * @param live
     * @param sseEmitter
     * @return
     * @param <T>
     */
    private static <T>boolean sendToClientCheck(Long userId,String live,SseUTF8 sseEmitter){
        //推送之前检测心跳是否存在
        boolean isAlive=checkSseConnectAlive(sseEmitter);
        if(!isAlive){
            //失去连接移除
            log.error("SSE推送消息失败：客户端{}未创建长连接或者关闭，失败信息：{}",userId,live);
            SSE_MAP_CACHE.remove(userId);
            return false;
        }
        SseEmitter.SseEventBuilder sendData=SseUTF8.event()
                .id(String.valueOf(HttpStatusEnum.SUCCESS.getCode()))
                .data(live,MediaType.APPLICATION_JSON);
        try {
            sseEmitter.send(sendData);
            return true;
        }catch(IOException e){
            log.error("SSE推送消息失败：用户{}，失败信息：{}",userId,live);
        }
        return true;
    }

    @Async
    @Scheduled(cron = "0 */30 * * * ?")
    public void checkSseAlive() {
        log.info("检测客户端连接状态");
        sendTextMessageCheck("LIVE");
    }


}
